package com.atguigu.app.dwd.db;

import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DwdLearnTestDetail
 * Package: com.atguigu.app.dwd
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/15 21:27
 * @Version 1.0
 */
public class DwdLearnExamDetail {
    public static void main(String[] args) {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2. 状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 3.使用FlinkSQL方式读取Kafka topic_db 主题数据
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_learn_test_detail"));

        //TODO 4.过滤出test_exam数据
        Table testExam = tableEnv.sqlQuery("" +
                "  select\n" +
                " `data`['id'] id,\n" +
                " `data`['paper_id'] paper_id,\n" +
                " `data`['user_id'] user_id,\n" +
                " `data`['score'] score,\n" +
                " `data`['duration_sec'] duration_sec,\n" +
                " `data`['create_time'] create_time,\n" +
                " `data`['submit_time'] submit_time,\n" +
                " `data`['update_time'] update_time,\n" +
                " `data`['deleted'] deleted,\n" +
                "    `pt`\n" +
                " from topic_db\n" +
                " where `database`='edu'\n" +
                " and `table`='test_exam'\n" +
                " and `type`='insert'");
        tableEnv.createTemporaryView("test_exam", testExam);

        //TODO 5.加载MySQL test_paper表作为LookUp表
        tableEnv.executeSql(MySqlUtil.getMysqlTestPaper());

        //TODO 6.关联
        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "  te.id,\n" +
                "  te.paper_id,\n" +
                "  te.user_id,\n" +
                "  te.score,\n" +
                "  te.duration_sec,\n" +
                "  te.create_time,\n" +
                "  te.submit_time,\n" +
                "  te.update_time,\n" +
                "  te.deleted,\n" +
                "  tp.paper_title,\n" +
                "  tp.course_id\n" +
                " from test_exam te\n" +
                " join test_paper \n" +
                " for system_time as of te.pt  as tp\n" +
                " on te.paper_id=tp.id");
        tableEnv.createTemporaryView("result_table",resultTable);

        //TODO 7.创建Kafka DWD层测验明细主题
        tableEnv.executeSql("" +
                "create table dwd_learn_exam_detail(\n" +
                "   id string,\n" +
                " paper_id string,\n" +
                " user_id string,\n" +
                " score string,\n" +
                " duration_sec string,\n" +
                " create_time string,\n" +
                " submit_time string,\n" +
                " update_time string,\n" +
                " deleted string," +
                " paper_title string,\n" +
                " course_id string\n" +
                ")"+KafkaUtil.getKafkaSinkConnOption("dwd_learn_exam_detail"));
        //TODO 8.写出数据
        tableEnv.executeSql("insert into dwd_learn_exam_detail select * from result_table");


    }

}
